/*
*    Author :huyn
*    2017/08/03
*    version 0.0.9
*    封装下层的方法供C++调用
*/
#include "htsc_grpc_trace_for_cc.h"

#include "stdio.h"
#include "htsc_grpc_utils.h"
#include "htsc_grpc_conf.h"
#include "htsc_grpc_properties_tools.h"
#include "htsc_grpc_common_utils.h"

static int htsc_grpc_trace_sender_threadcount = -1;
static htsc_grpc_trace_link_head_t htsc_grpc_tracelinks[HTSC_GRPC_TRACE_READ_THREADCOUNT];

//-------------------------------------------------------------
//通过数组方式读写服务跟踪信息
htsc_grpc_messageinfo_t *htsc_grpc_trace_array[HTSC_GRPC_TRACE_READ_THREADCOUNT];
static long htsc_grpc_trace_index_write[HTSC_GRPC_TRACE_READ_THREADCOUNT];
static long htsc_grpc_trace_index_read[HTSC_GRPC_TRACE_READ_THREADCOUNT];
static long htsc_grpc_trace_array_len = 10000;
static char * htsc_grpc_trace_message[HTSC_GRPC_TRACE_READ_THREADCOUNT];
gpr_mu htsc_grpc_array_index_mu[HTSC_GRPC_TRACE_READ_THREADCOUNT];
static size_t htsc_grpc_trace_message_size_mul = sizeof(char) *HTSC_GRPC_TRACE_MESSAGE_CHARS_MULTIPLE;
static size_t htsc_grpc_trace_message_size_sin = sizeof(char) *HTSC_GRPC_TRACE_MESSAGE_CHARS_SINGLE;

//根据htsc_grpc_trace_sender_threadcount数组分配数组长度

//初始化操作
void htsc_grpc_trace_array_init() {
	int i = 0;
	int j = 0;
	for (i = 0;i < htsc_grpc_trace_sender_threadcount;i++) {
		gpr_mu_init(&htsc_grpc_array_index_mu[i]);
		htsc_grpc_trace_array[i] = (htsc_grpc_messageinfo_t*) malloc(htsc_grpc_trace_array_len * sizeof(htsc_grpc_messageinfo_t));
		for (j = 0;j < htsc_grpc_trace_array_len;j++) {
			htsc_grpc_trace_array[i][j].message = NULL;
			htsc_grpc_trace_array[i][j].record_state = 0;
		}
		htsc_grpc_trace_index_write[i] = -1;
		htsc_grpc_trace_index_read[i] = -1;
		htsc_grpc_trace_message[i] = (char*)malloc(htsc_grpc_trace_message_size_mul);
	}
}

//获取指定数组的写下标
long htsc_grpc_trace_new_write_index(int array_index) {
	long index = 0;
	//定义多个锁是否执行速度更快  
	gpr_mu_lock(&htsc_grpc_array_index_mu[array_index]);
	htsc_grpc_trace_index_write[array_index] = (htsc_grpc_trace_index_write[array_index] + 1) % htsc_grpc_trace_array_len;
	index = htsc_grpc_trace_index_write[array_index];
	gpr_mu_unlock(&htsc_grpc_array_index_mu[array_index]);
	return index;
}

//写数组 
void htsc_grpc_trace_array_write(int array_index, char *message) {
	long write_index = htsc_grpc_trace_new_write_index(array_index);
	//分配空间 
	if (htsc_grpc_trace_array[array_index][write_index].message == NULL) {
		htsc_grpc_trace_array[array_index][write_index].message = (char*)malloc(htsc_grpc_trace_message_size_sin);
	}
	/*else {
		printf("\n ----------------old message:%s", htsc_grpc_trace_array[array_index][write_index].message);
	}*/
	//清空内容写入message
	//int len()
	memset(htsc_grpc_trace_array[array_index][write_index].message, 0, htsc_grpc_trace_message_size_sin);
	strcpy(htsc_grpc_trace_array[array_index][write_index].message,message);
	htsc_grpc_trace_array[array_index][write_index].record_state = 1;
}

//读数组
long htsc_grpc_message_read_count=0;
char * htsc_grpc_trace_array_read(int array_index) {
	long read_index = (htsc_grpc_trace_index_read[array_index] + 1) % htsc_grpc_trace_array_len;
	int rec_state = htsc_grpc_trace_array[array_index][read_index].record_state;
	memset(htsc_grpc_trace_message[array_index], 0, htsc_grpc_trace_message_size_mul);
	strcpy(htsc_grpc_trace_message[array_index], "[");
	while (rec_state == 1) {
		//如果超过字符数限制，则返回。
		if ((HTSC_GRPC_TRACE_MESSAGE_CHARS_MULTIPLE - strlen(htsc_grpc_trace_message[array_index])) < HTSC_GRPC_TRACE_MESSAGE_CHARS_SINGLE) {
			break;
		}
		//huyntodo ---------------test comment
		/*htsc_grpc_message_read_count++;
		if (htsc_grpc_message_read_count > 10000000) {
			htsc_grpc_message_read_count = 0;
		}*/
		/*printf("\n--------------------------readmessage total Count:%d",htsc_grpc_message_read_count);*/
		htsc_grpc_trace_index_read[array_index]++;
		strcat(htsc_grpc_trace_message[array_index], htsc_grpc_trace_array[array_index][read_index].message);
		htsc_grpc_trace_array[array_index][read_index].record_state = 0;
		read_index = (htsc_grpc_trace_index_read[array_index] + 1) % htsc_grpc_trace_array_len;
		rec_state = htsc_grpc_trace_array[array_index][read_index].record_state;
	}
	if (strcmp(htsc_grpc_trace_message[array_index], "[") == 0) {
		return  NULL;
	}
	else {
		strcat(htsc_grpc_trace_message[array_index], "]");
	}
	return htsc_grpc_trace_message[array_index];
}
//--------------------------------------------------------------
 

/*
* 获取发送线程数
*/
int htsc_grpc_trace_getsendthreadcount() {
	return htsc_grpc_trace_sender_threadcount;
}


//初始化队列数组
void htsc_grpc_inittracesender() {	
	size_t buffsize = sizeof(char) * 100;
	char *threadcount = (char*)malloc(buffsize);
	memset(threadcount, 0, buffsize);
	htsc_grpc_properties_get_value(HTSC_GRPC_CONF_KAFKA_SENDER_NUMBER, NULL, threadcount);
	//如果为配置启动一个线程来反思信息
	if (threadcount == NULL || strlen(threadcount) == 0) { 
		htsc_grpc_trace_sender_threadcount = 1;
	}
	else if(htsc_grpc_common_utils_isdigit(threadcount)==false){
		htsc_grpc_trace_sender_threadcount = atoi(threadcount);
	}
	else {
		htsc_grpc_trace_sender_threadcount = 1;
	}
	//配置的线程数不允许超过HTSC_GRPC_TRACE_READ_THREADCOUNT
	if (htsc_grpc_trace_sender_threadcount >HTSC_GRPC_TRACE_READ_THREADCOUNT) {
		htsc_grpc_trace_sender_threadcount = HTSC_GRPC_TRACE_READ_THREADCOUNT;
	}
	htsc_grpc_trace_array_init();
	/*int count = htsc_grpc_trace_sender_threadcount;
	for (int i = 0;i < count;i++) {
		htsc_grpc_trace_link_head_t tracenode = { .ptfirst = NULL,.ptread = NULL,.ptend = NULL };
		htsc_grpc_tracelinks[i] = tracenode;
	}*/
}

/*
* 返回指定头结点指针
*/
htsc_grpc_trace_link_head_t *htsc_grpc_gettracehead_pt(int index) {
	return &htsc_grpc_tracelinks[index];
}

/*
* 释放htsc_grpc_trace_link_node_t节点信息
*/
void htsc_grpc_link_node_free(htsc_grpc_trace_link_node_t **tracenode) {
	htsc_grpc_trace_link_node_t *ptnode = *tracenode;
	ptnode->traceinfo = NULL; //traceinfo内容在别处进行释放
	ptnode->next = NULL;
	free(ptnode);
}

void htsc_grpc_trace_item_cat(char *outjson,char *pref,char *key,char *value) {
	strcat(outjson, pref);
	strcat(outjson, "\"");
	strcat(outjson, key);
	strcat(outjson, "\":\"");
	if (value == NULL) {
		strcat(outjson, "");
	}
	else {
		strcat(outjson, value);
	}	
	strcat(outjson, "\"");
}

void htsc_grpc_trace_to_jsonstr(hstc_grpc_common_traceinfo_t *traceinfo,char *outjson) {	
	strcpy(outjson, "{");	 
	char buf[100];
	htsc_grpc_trace_item_cat(outjson, "",  "traceId", traceinfo->traceid);
	htsc_grpc_trace_item_cat(outjson, ",", "chainId", traceinfo->chainid);
	htsc_grpc_trace_item_cat(outjson, ",", "initial", traceinfo->initial?"true":"false");
	htsc_grpc_trace_item_cat(outjson, ",", "serviceName", traceinfo->servicename);
	htsc_grpc_trace_item_cat(outjson, ",", "methodName", traceinfo->methodname);
	REINIT(buf);
	sprintf(buf, "%ju", traceinfo->starttime);
	htsc_grpc_trace_item_cat(outjson, ",", "startTime", gprc_strdup(buf));
	REINIT(buf);
	sprintf(buf, "%ju", traceinfo->endtime);
	htsc_grpc_trace_item_cat(outjson, ",", "endTime", gprc_strdup(buf));
	htsc_grpc_trace_item_cat(outjson, ",", "success", traceinfo->success ? "true" : "false");
	htsc_grpc_trace_item_cat(outjson, ",", "consumerSide", traceinfo->consumerside ? "true" : "false");
	htsc_grpc_trace_item_cat(outjson, ",", "consumerHost", traceinfo->consumerhost);
	REINIT(buf);
	sprintf(buf, "%d", traceinfo->consumerport);
	htsc_grpc_trace_item_cat(outjson, ",", "consumerPort", gprc_strdup(buf));
	htsc_grpc_trace_item_cat(outjson, ",", "providerHost", traceinfo->providerhost);
	REINIT(buf);
	sprintf(buf, "%d", traceinfo->providerport);
	htsc_grpc_trace_item_cat(outjson, ",", "providerPort", gprc_strdup(buf));
	htsc_grpc_trace_item_cat(outjson, ",", "protocol", traceinfo->protocol);
	htsc_grpc_trace_item_cat(outjson, ",", "appName", traceinfo->appname);
	htsc_grpc_trace_item_cat(outjson, ",", "serviceGroup", traceinfo->servicegroup);
	htsc_grpc_trace_item_cat(outjson, ",", "serviceVersion", traceinfo->serviceversion);
	if (traceinfo->pushtime > 0) {		
		REINIT(buf);
		sprintf(buf, "%ld", traceinfo->pushtime);
		htsc_grpc_trace_item_cat(outjson, ",", "pushTimes", gprc_strdup(buf));
	}
	strcat(outjson, "}");
}

void htsc_grpc_cat_trace_jsonstr(char *message, char *outjson) {
	if (strcmp(outjson, "[") != 0) { 
		strcat(outjson, ",");		
	}
	strcat(outjson, message);
}

//读取链表节点，读取节点后删除节点，如果被读取节点最后一个节点则不删除，修改读取标记为已读取
int readcount = 0;
char *htsc_grpc_trace_link_read(int index) {
	return htsc_grpc_trace_array_read(index);
	//return NULL;
}

char * htsc_grpc_trace_link_read_bk(int index) {
	htsc_grpc_trace_link_head_t *pt = htsc_grpc_gettracehead_pt(index);
	//链表不包含节点
	if (pt->isfirst != 1) {
		return NULL;
	}
	//如果pt->ptread == NULL说明是未读取取过，从首节点开始读取。
	if (pt->ptread == NULL) {
		pt->ptread = pt->ptfirst;
		pt->ptfirst = NULL;
	}

	if (pt->ptread == NULL) {
		printf(" \n---Error:trace link is null.");
		return NULL;
	}
	int maxcharnumbers = 100000;
	size_t size = maxcharnumbers * sizeof(char);
	char *result = (char*)malloc(size);
	memset(result, 0, size);
	strcpy(result, "[");

	htsc_grpc_trace_link_node_t *ptnode = pt->ptread;
	htsc_grpc_trace_link_node_t *ptnext = NULL;
	while (ptnode != NULL) {
		//分配内存空间少于500字符是不在读取。
		if ((maxcharnumbers - strlen(result)) < 500) {
			break;
		}
		ptnext = ptnode->next;
		//拼接字符串
		//如果不包含下级节点，则不删除本节点之间修改标记
		if (ptnext != NULL && ptnext->readflag != 0 && ptnext->readflag != 1) {
			printf("\n==============================readDataError!");
		}
		if (ptnode->readflag != 0 && ptnode->readflag != 1) {
			printf("\n==============================readDataError!");
		}
		if (ptnext == NULL) {
			if (ptnode->readflag == 1) {
				continue;
			}
			htsc_grpc_cat_trace_jsonstr(ptnode->traceinfo, result);
			readcount++;
			ptnode->readflag = 1;
			pt->ptread = ptnode;
		}
		else {
			if (ptnode->readflag == 1) {

			}
			else {
				htsc_grpc_cat_trace_jsonstr(ptnode->traceinfo, result);
				readcount++;
			}
			pt->ptread = ptnext;
			//删除ptnode节点
			ptnode->next = NULL;
			htsc_grpc_link_node_free(&ptnode);
		}
		ptnode = ptnext;
	}
	if (strcmp(result, "[") == 0) {
		result = NULL;
	}
	else {
		strcat(result, "]");
	}
	//printf("\n--------------total readcount: %d", readcount);
	return result;
}

//把服务跟踪信息写入链表，写入时需要加锁
void htsc_grpc_trace_write(hstc_grpc_common_traceinfo_t *traceinfo) {
	//return;
	//获取队列数组下标
	int index = traceinfo->endtime % htsc_grpc_trace_sender_threadcount;
 
 
	size_t size = HTSC_GRPC_TRACE_MESSAGE_CHARS_SINGLE * sizeof(char);
	char *message = (char*)malloc(size);
	memset(message, 0, size);
	htsc_grpc_trace_to_jsonstr(traceinfo, message);
	//printf("\n--------------------------write message:\n %s\n", message);
	htsc_grpc_trace_array_write(index, message);
	free(message);
	//原来写链表功能（作废）
	//htsc_grpc_trace_write_link(message);
}

void htsc_grpc_trace_write_link(int index,char *message) {
	htsc_grpc_trace_link_head_t *pt = htsc_grpc_gettracehead_pt(index);
	htsc_grpc_trace_link_node_t *ptnode = (htsc_grpc_trace_link_node_t*)malloc(sizeof(htsc_grpc_trace_link_node_t));
	ptnode->readflag = 0;
	ptnode->traceinfo = message;
	ptnode->next = NULL;

	//lock 
	if (pt->isfirst == 0) {
		gpr_mu_init(&pt->mu);
	}
	gpr_mu_lock(&pt->mu);
	if (pt->isfirst == 0) {
		pt->ptfirst = ptnode;
		pt->ptend = ptnode;
		pt->isfirst = 1;
	}
	else {
		pt->ptend->next = ptnode;
		pt->ptend = ptnode;
	}
	gpr_mu_unlock(&pt->mu);
}

/*
*func：获取kafka代理服务器地址
*date：20170811 
*code：huyn
*/
void htsc_grpc_trace_read_kafkaproxy_servers(char *confitem) {
	size_t buffsize = 500 * sizeof(char);
	confitem = (char*)malloc(buffsize);
	memset(confitem, 0, buffsize);
	htsc_grpc_properties_get_value(HTSC_GRPC_CONF_KAFKA_SENDER_SERVERS, NULL, confitem);
}


//返回读取记录条数
long htsc_grpc_trace_read_count() {
	return htsc_grpc_message_read_count;
}





